//
// Created by YuWei on 2023/10/24.
//

#include "common/log/log.h"
#include "sql/operator/update_physical_operator.h"
#include "storage/record/record.h"
#include "storage/table/table.h"
#include "storage/trx/trx.h"
#include "sql/stmt/update_stmt.h"

RC UpdatePhysicalOperator::open(Trx *trx)
{
  if(children_.empty()){
    return RC::SUCCESS;
  }

  std::unique_ptr<PhysicalOperator> &child = children_[0];
  RC rc = child->open(trx);
  if(rc != RC::SUCCESS){
    LOG_WARN("failed to open child operator: %s", strrc(rc));
    return rc;
  }

  trx_ = trx;
  return RC::SUCCESS;
}

RC UpdatePhysicalOperator::next()
{
  RC rc = RC::SUCCESS;
  if(children_.empty()){
    return RC::RECORD_EOF;
  }
  PhysicalOperator *child = children_[0].get();
  int updated_count = 0;
  while(RC::SUCCESS == (rc = child->next())){
    Tuple *tuple = child->current_tuple();
    if(nullptr == tuple){
      LOG_WARN("failed to get current record: %s", strrc(rc));
      return rc;
    }
    RowTuple *row_tuple = static_cast<RowTuple *>(tuple);
    Record &record = row_tuple->record();
    for(int i=0;i<fields_.size();i++){
      PhysicalOperator* sub_oper=sub_opers_[i].get();
      if(sub_oper!=nullptr){
      sub_oper->open(trx_);
      sub_oper->add_p_tuple(tuple);
      rc=sub_oper->next();
      if(rc!=RC::SUCCESS){
        if(fields_[i]->can_be_null()){
         Value sub_value= values_[i];
         sub_value.set_null();
         values_[i]=sub_value;
          sub_oper->pop_p_tuple();
          sub_oper->close();
          continue;
        }
        LOG_WARN("failed to get sub_tuple in update");
        return RC::INVALID_ARGUMENT;
      }
      Tuple* sub_tuple=sub_oper->current_tuple();
      if(sub_tuple==nullptr||sub_tuple->cell_num()!=1){
        LOG_WARN("sub_tuple error in update");
        return RC::INVALID_ARGUMENT;
      }
      Value sub_value;
      sub_tuple->cell_at(0,sub_value);
      if(sub_oper->next()!=RC::RECORD_EOF){
         LOG_WARN("multy sub_tuple error in update");
        return RC::INVALID_ARGUMENT;
      }
      sub_value.cast_to_type(fields_[i]->type());
      values_[i]=sub_value;
      sub_oper->pop_p_tuple();
      sub_oper->close();
      }
    }
    Record new_record;
    table_->make_new_record(record,new_record,fields_,values_);
    rc = trx_->update_record(table_,record,new_record);
    if (rc != RC::SUCCESS) {
      LOG_WARN("failed to delete record: %s", strrc(rc));
      return rc;
    }
    updated_count++;
  }
  return RC::RECORD_EOF;
}

RC UpdatePhysicalOperator::close()
{
  if (!children_.empty()) {
    children_[0]->close();
  }
  for(int i=0;i<sub_opers_.size();i++){
    PhysicalOperator* sub_oper=sub_opers_[i].get();
    if(sub_oper!=nullptr){
      sub_oper->close();
    }
  }
  return RC::SUCCESS;
}